package standalone.sna.community;

import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;

import cern.colt.list.DoubleArrayList;
import cern.colt.list.IntArrayList;
import cern.colt.matrix.DoubleFactory1D;
import cern.colt.matrix.DoubleMatrix1D;
import cern.colt.matrix.DoubleMatrix2D;
import reducible.VectorReducible;
import utils.DataInput;
import utils.MathUtils;
import allreduce.AllReducer;

/**
 * Community Detection by *Hard* Label Propagation
 * @author purlin
 *
 */
public class LabelPropagation {
	/**
	 * Number of Users
	 */
	protected int numUsers;
	/**
	 * Number of communities
	 */
	protected int numCommunities;
	/**
	 * Model, cluster index
	 */
	protected VectorReducible w;
	/**
	 * Allreduce object for communication
	 */
	protected AllReducer reduceObj;
	/**
	 * Initialization
	 * @param master: master IP
	 * @param masterPort: port of master
	 * @throws IOException
	 */
	public void initialize(String master, int masterPort) throws IOException {
		//Initialize w as zero vector
		w = new VectorReducible(DoubleFactory1D.sparse.make(numUsers));
		//Initialize allreduce object;
		reduceObj = new AllReducer(master, masterPort);
		reduceObj.init();
	}
	/**
	 * Label propagation
	 * @param neighborList
	 * @param numIt
	 * @throws IOException
	 */
	public void performLP(DoubleMatrix2D neighborList, List<Integer> idList, int numIt) throws IOException {
		//Community index Initialization 
		for(Integer id: idList) w.vectors.set(id, Math.floor((Math.random()*numCommunities)));
		w = (VectorReducible) reduceObj.run(w);
		//Data parameters
		IntArrayList cidList = new IntArrayList();
		DoubleArrayList cvalueList = new DoubleArrayList();
		//Iteration
		for(int i=0;i<numIt;i++){
			DoubleMatrix1D currentIDs = DoubleFactory1D.sparse.make(numUsers);
			//Get neighbors' labels
			for(int j=0;j<neighborList.rows();j++){
				int[] comCount = new int[numCommunities];
				neighborList.viewRow(j).getNonZeros(cidList, cvalueList);
				for(int k=0;k<cidList.size();k++) comCount[(int)cvalueList.get(k)] += 1;
				currentIDs.set(idList.get(j), MathUtils.getMaxID(comCount));
			}
			VectorReducible cw = new VectorReducible(currentIDs);
			w = (VectorReducible) reduceObj.run(cw);
		}
	}
	
	/**
	 * Save model
	 * @param out: output stream
	 * @throws IOException
	 */
	public void saveModel(OutputStream out) throws IOException{
		for(int i=0;i<numUsers;i++)
			out.write((this.w.vectors.get(i)+"\n").getBytes());
	}
	
	/**
	 * The rank of the current client
	 * @return
	 */
	public int getRank() {
		return this.reduceObj.rank;
	}
	
	/**
	 * Constructor
	 * @param numUsers: number of users
	 * @param numCommunities: number of communities
	 */
	public LabelPropagation(int numUsers, int numCommunities){
		this.numCommunities = numCommunities;
		this.numUsers = numUsers;
	}
	
	/**
	 * Build Model
	 * @param inputPath: input path
	 * @param outputPath: output path for network indicators
	 * @param dataType: data format: pair list or link table
	 * @param modelName: name of model to save
	 * @param numUsers: number of users:
	 * @param numCommunities: number of communities
	 * @param numIt: number of iterations
	 * @param master: master IP
	 * @param masterPort: master port
	 * @throws NumberFormatException
	 * @throws IOException
	 */
	public static void buildModel(String inputPath, String outputPath, String dataType, String modelName, int numUsers, int numCommunities, int numIt, String master, int masterPort) throws NumberFormatException, IOException {
		//Read data
		List<Integer> idList = new ArrayList<Integer>();
		DoubleMatrix2D instances = null;
		if("pair".endsWith(dataType)) instances = DataInput.readPairList(inputPath, numUsers, idList);
		else if("link".endsWith(dataType)) instances = DataInput.readDenseMatrix(inputPath, numUsers, idList);
		//Model building
		LabelPropagation lpObj = new LabelPropagation(numUsers, numCommunities);
		lpObj.initialize(master, masterPort);
		lpObj.performLP(instances, idList, numIt);
		//Model saving
		if(lpObj.getRank()==0){
			DataOutputStream out = new DataOutputStream(new FileOutputStream(new File(outputPath+"/"+modelName)));
			lpObj.saveModel(out);
			out.close();
		}
	}
}
